package com.sxzjrj.cmcc

import com.alibaba.fastjson.{JSON, JSONObject}
import com.sxzjrj.bean.CmccBean
import com.sxzjrj.config.CmccConfig
import com.sxzjrj.constant.Constant
import com.sxzjrj.utils.{CmccUtils, JdbcUtils}
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}

/**
  * Created by ljj in 2019/3/20
  *
  */
object CmccSparkSql {

  def main(args: Array[String]): Unit = {

    val session = SparkSession
      .builder()
      .appName("充值平台监控")
      .master("local[*]")
      .getOrCreate()


    val file = session.read.textFile("e:/cmcc.log")

    import session.implicits._
    val data = file.filter(_.contains(Constant.RECEIVENOTIFYTIME.toString))
      .map(line => {

        var province: String = "000"
        var money: Int = 0
        var time: Long = 0L
        var succOrder: Int = 0
        val order = 1

        try {

          val bean: JSONObject = JSON.parseObject(line)
          val bussinessRst = bean.getString("bussinessRst")
          val flag = bussinessRst.equals("0000")

          val costTime = CmccUtils.getCostTime(bean.getString("requestId"), bean.getString("receiveNotifyTime"))

          province = if (bean.getString("provinceCode").length > 0) bean.getString("provinceCode") else "000"

          succOrder = if (flag) 1 else 0

          time = if (flag) costTime else 0L

          money = if (flag) {
            if (bean.getString("chargefee").length > 0) bean.getString("chargefee").toInt else 0
          } else 0
        } catch {
          case e: Exception => e.printStackTrace()
        }
        (province, succOrder, money, if (time >= 0L) time else time * (-1))

      }).toDF("province","succOrder","chargefee","time")
    data.cache()
    data.createTempView("t_province")

    val result = session.sql(
      """
        |select province,count(*) as order,sum(succOrder) as succOrderSum, sum(chargefee) as money,sum(time) as timeSum
        |from t_province group by province
      """.stripMargin)

    result.write.mode(SaveMode.Append)
//      //.text("F:/cmcc")
      //.json("F:/cmcc/cmcc.txt")
      .jdbc(CmccConfig.url,"t_cmcc_core",CmccConfig.properties)
//
//    //JdbcUtils.getData4Mysql

    session.stop()


  }

}
